package cn.z2huo.demo.kafka.apache.configtest.producer;

import cn.z2huo.demo.kafka.apache.jackson.KafkaJsonMapper;
import cn.z2huo.demo.kafka.apache.producer.AsyncSendMessageCallback;
import cn.z2huo.demo.kafka.apache.producer.ProducerProperties;
import cn.z2huo.demo.model.dataobject.user.UserDO;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.json.JsonMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Properties;

/**
 * <p>异步发送消息的生产者
 *
 * @author z2huo
 */
@Slf4j
public class UserProducer {

    private static final String TOPIC = "test-topic";

    /**
     * 生成的用户数量
     */
    private static final int GENERATE_USER_COUNT = 1;

    public static void main(String[] args) {
        new UserProducer().send(GENERATE_USER_COUNT);
    }

    public void send(int count) {

        long start, end;

        Properties props = ProducerProperties.getProducerProperties();

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {

            List<UserDO> userDOS = DataGeneratorUtils.generateUser(count);
            JsonMapper jsonMapper = KafkaJsonMapper.INSTANCE;

            start = System.currentTimeMillis();

            long totalBytes = 0L;
            for (int i = 0; i < userDOS.size(); i++) {
                log.info("send the NO.{} user", i);
                String userDOJsonString = jsonMapper.writeValueAsString(userDOS.get(i));
                int bytesLength = userDOJsonString.getBytes(StandardCharsets.UTF_8).length;
                totalBytes += bytesLength;
                log.info("userDO json String length is {}, size is {} bytes, userDOJsonString is {}", userDOJsonString.length(),
                        bytesLength, userDOJsonString);
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, userDOJsonString);
                producer.send(record, new AsyncSendMessageCallback());
            }
            // 刷新并关闭生产者
            producer.flush();

            end = System.currentTimeMillis();
            log.info("send {} users cost {} ms, all json string bytes size is {}", count, end - start, totalBytes);
        } catch (JsonProcessingException e) {
            log.error("write user to json fail");
            throw new RuntimeException(e);
        }
    }

}
